package org.my.hadoop.test;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class WordCount {
   public static class Map extends  Mapper<Object,Text,Text,IntWritable>{
    
   private final static IntWritable one = new IntWritable(1);
   private Text word = new Text();
   
   public void map(Object key,Text value,Context context) throws IOException, InterruptedException{
	   String line = value.toString();
	   StringTokenizer tokenizer = new StringTokenizer(line);
	   while(tokenizer.hasMoreTokens()){
		   word.set(tokenizer.nextToken());
		   context.write(word, one);
	   }
   }
  }
  
   
  public static class Reduce  extends Reducer<Text, IntWritable, Text, IntWritable>{
	  private IntWritable result = new IntWritable();
	  public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException{
		int sum = 0;
		for(IntWritable val:values){
			sum += val.get();
		}
		result.set(sum);
		context.write(key,result);
	}
  } 
  
  
  public static void main(String[] args) throws Exception{
	  
	  Configuration configuration =new Configuration();
	  Job job = Job.getInstance(configuration,"word count");
	  job.setJarByClass(WordCount.class);

	  job.setMapperClass(Map.class);
	  job.setCombinerClass(Reduce.class);
	  job.setReducerClass(Reduce.class);
	  
	  job.setOutputKeyClass(Text.class);
	  job.setOutputValueClass(IntWritable.class);
	  
	  FileInputFormat.addInputPath(job, new Path(args[0]));
	  FileOutputFormat.setOutputPath(job, new Path(args[1]));
	  
	  System.exit(job.waitForCompletion(true)?0:1);
  }
}
